package com.shujia.flink.tf

import org.apache.flink.streaming.api.TimerService
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo12KeyByProcess {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    val kvDS: DataStream[(String, Int)] = linesDS.flatMap(_.split(",")).map((_, 1))

    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)

    /**
     * KeyedProcessFunction: 在keyBy之后使用的算子，在使用有状态计算是，用一个key会保存一个状态
     */

    keyByDS.process(new KeyedProcessFunction[String, (String, Int), (String, Int)] {
      /**
       * processElement： 每一条数据执行一次
       *
       * @param value ：一条数据
       * @param ctx   ：上下文对象
       * @param out   ：用于将数据发送到下游
       */
      override def processElement(value: (String, Int),
                                  ctx: KeyedProcessFunction[String, (String, Int), (String, Int)]#Context,
                                  out: Collector[(String, Int)]): Unit = {
        println(value)

        //定时器对象
        val timerService: TimerService = ctx.timerService()

        //获取当前处理的时间
        val currentTime: Long = timerService.currentProcessingTime()
        println(currentTime)
        //设置定期器， 在5秒之后触发onTimer执行
        timerService.registerProcessingTimeTimer(currentTime + 5000)
      }

      /**
       * onTimer: 定时器触发的方法
       *
       * @param timestamp ：时间
       * @param ctx       ：上下文读写
       * @param out       ：用于将数据发送到下游
       */
      override def onTimer(timestamp: Long,
                           ctx: KeyedProcessFunction[String, (String, Int), (String, Int)]#OnTimerContext,
                           out: Collector[(String, Int)]): Unit = {

        println(s"onTimer执行了：$timestamp")
      }
    })

    env.execute()
  }

}
